/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package dsmsforkjoin.joins;

import dsmsforkjoin.FJoin;
import java.util.ArrayList;
import tuplegenerator.Tuple;

/**
 *
 *
 */
public class SortMergeJoinOperator<T1 extends Tuple, T2 extends Tuple> implements IJoinOperator<T1, T2> {

    @Override
    public void doJoin(FJoin<T1> fj1, FJoin<T2> fj2) {
        ArrayList<Tuple> tp1 = fj1.generate();
        ArrayList<Tuple> tp2 = fj2.generate();
        
        int i = 0, j = 0;
        
        Tuple t1 = tp1.get(i);
        Tuple t2 = tp2.get(j);
        
        while (i < tp1.size() && j < tp2.size()) {
            
            if (t1.getKey() < t2.getKey()) {
                i++;
                t1 = tp1.get(i);
            } else if (t1.getKey() > t2.getKey()) {
                j++;
                t2 = tp2.get(j);
            } else {
                //System.out.println(tp1.get(i).toString() + "==>" + tp2.get(j).toString());
                j++;
                if (j >= tp2.size()) break;
                t2 = tp2.get(j);
            }
        }
        
    }
    
}
